package com.cookie.utils;

import com.alibaba.fastjson.JSONObject;
import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.protocol.CanalEntry.Column;
import com.alibaba.otter.canal.protocol.CanalEntry.Entry;
import com.alibaba.otter.canal.protocol.CanalEntry.EntryType;
import com.alibaba.otter.canal.protocol.CanalEntry.EventType;
import com.alibaba.otter.canal.protocol.CanalEntry.RowChange;
import com.alibaba.otter.canal.protocol.CanalEntry.RowData;
import com.alibaba.otter.canal.protocol.Message;
import com.cookie.config.CanalClientConfig;
import com.cookie.config.RedisConfig;
import redis.clients.jedis.HostAndPort;

import java.net.InetSocketAddress;
import java.util.HashSet;
import java.util.List;
import java.util.Set;

public class CanalUtil {
    private String zkServers;
    private String ip;
    private int port;
    private String destination;
    private String filter = ".*\\..*";

    private RedisConfig redisConf;
    // 统一调用redis util来处理redis相关业务
    private RedisUtil redis;

    public CanalUtil(CanalClientConfig canalConf, RedisConfig redisConf){
        if (canalConf.getZkServers()!=null){
            zkServers = canalConf.getZkServers();
        }
        // System.out.println(zkServers);
        ip = canalConf.getIp();
        port = canalConf.getPort();
        destination = canalConf.getDestination();
        filter = canalConf.getFilter();
        this.redisConf = redisConf;
    }
    public void start(){
        // 根据配置连接单机或集群模式的canal
        CanalConnector connector = null;
        if(zkServers==null){
             connector = CanalConnectors.newSingleConnector(
                    new InetSocketAddress(ip,port), destination, "", "");
            System.out.println("单机模式下canal连接成功");
        }else {
            connector = CanalConnectors.newClusterConnector(
                    zkServers,destination, "", "");
            System.out.println("zk模式下canal连接成功");
        }

        // 创建redis工具类实例
        redis = new RedisUtil(redisConf);

        int batchSize = 1000;
        int emptyCount = 0;

        try {
            connector.connect();
            // 该方法会更新过滤规则
            connector.subscribe(filter);
            connector.rollback();
            int totalEmptyCount = 120;
            // 关掉了测试环境下的自动退出机制
            // while (emptyCount < totalEmptyCount) {
            while (true){
                Message message = connector.getWithoutAck(batchSize); // 获取指定数量的数据
                long batchId = message.getId();
                int size = message.getEntries().size();
                if (batchId == -1 || size == 0) {
                    emptyCount++;
                    // System.out.println("empty count : " + emptyCount);
                    try {
                        Thread.sleep(1000);
                    } catch (InterruptedException e) {
                    }
                } else {
                    emptyCount = 0;
                    // System.out.printf("message[batchId=%s,size=%s] \n", batchId, size);
                    printEntry(message.getEntries());
                }

                connector.ack(batchId); // 提交确认
                // connector.rollback(batchId); // 处理失败, 回滚数据
            }

            // System.out.println("empty too many times, exit");
        } finally {
            connector.disconnect();
        }
    }

    private void printEntry(List<Entry> entrys) {
        for (Entry entry : entrys) {
            if (entry.getEntryType() == EntryType.TRANSACTIONBEGIN || entry.getEntryType() == EntryType.TRANSACTIONEND) {
                continue;
            }

            RowChange rowChage = null;
            try {
                rowChage = RowChange.parseFrom(entry.getStoreValue());
            } catch (Exception e) {
                throw new RuntimeException("ERROR ## parser of eromanga-event has an error , data:" + entry.toString(), e);
            }

            EventType eventType = rowChage.getEventType();
            // 使用表名作为hash的key, id作为hash的field, 各个字段组成json字符串作为value(不是
            // 使用 表名:id 作为key, 各字段组成的json字符串作为value, 存储为redis的string类型(旧
            // 使用 prefix:table:id 作为key, 各字段和字段值作为field和value
            String tableName = entry.getHeader().getTableName();
            System.out.println(String.format("================&gt; binlog[%s:%s] , name[%s,%s] , eventType : %s",
                    entry.getHeader().getLogfileName(), entry.getHeader().getLogfileOffset(),
                    entry.getHeader().getSchemaName(), tableName, eventType));
            long start = System.currentTimeMillis();
            // 这里拿到的是一行数据，各个字段以及各字段的值
            for (RowData rowData : rowChage.getRowDatasList()) {
                // 如果是delete操作 就打印变更前的数据
                if (eventType == EventType.DELETE) {
                    redisDelete(tableName, rowData.getBeforeColumnsList());  // redis delete
                    printColumn(rowData.getBeforeColumnsList());
                    // 如果是插入操作，就打印变更后的数据
                } else if (eventType == EventType.INSERT) {
                    redisInsert(tableName, rowData.getAfterColumnsList());  // redis insert
                    printColumn(rowData.getAfterColumnsList());
                    // 如果是更新操作，则变更前后的数据都打印
                } else {
                    System.out.println("-------&gt; before");
                    printColumn(rowData.getBeforeColumnsList());
                    System.out.println("-------&gt; after");
                    redisInsert(tableName, rowData.getAfterColumnsList());  // redis update
                    printColumn(rowData.getAfterColumnsList());
                }
            }
            System.out.printf("sync cost: %d\n", System.currentTimeMillis()-start);
        }
    }

    private void printColumn(List<Column> columns) {
        for (Column column : columns) {
            System.out.println(column.getName() + " : " + column.getValue() + "    update=" + column.getUpdated());
        }
    }

    // redis的插入，删除，更新操作
    private void redisInsert(String tableName, List<Column> columns){
        JSONObject json=new JSONObject();
        for (Column column : columns) {
            json.put(column.getName(), column.getValue());
            // 插入hash类型
            // redis.hashSet(tableName+":"+columns.get(0).getValue(),column.getName(), column.getValue());
        }
        if(columns.size()>0){
            // redis.hashSet(tableName,columns.get(0).getValue(),json.toJSONString());
            // 插入string类型
            redis.stringSet(tableName+":"+ columns.get(0).getValue(),json.toJSONString());
        }
    }
    // 这个update方法和insert方法是一样的
    /*
    private void redisUpdate( List<Column> columns){
        JSONObject json=new JSONObject();
        for (Column column : columns) {
            json.put(column.getName(), column.getValue());
        }
        if(columns.size()>0){
            redis.stringSet("user:"+ columns.get(0).getValue(),json.toJSONString());
        }
    }*/
    private void redisDelete(String tableName, List<Column> columns){

        if(columns.size()>0){
            redis.delKey(tableName+":"+columns.get(0).getValue());
        }
    }
}
